package com.hngy.java.stream.transformaction;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Arrays;

/**
 * 使用sideoutput切分流
 */
public class StreamSideOutputJava {

    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Integer> text = env.fromCollection(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
        //按照数据的奇偶性对数据进行分流
        //首先定义两个sideoutput来准备保存切分出来的数据
        OutputTag<Integer> outputTag1 = new OutputTag<Integer>("even"){};
        OutputTag<Integer> outputTag2 = new OutputTag<Integer>("odd"){};

        //一次切分
        SingleOutputStreamOperator<Integer> outputStream = text.process(new ProcessFunction<Integer, Integer>() {
            @Override
            public void processElement(Integer value, Context ctx, Collector<Integer> out) throws Exception {
                if (value % 2 == 0) {
                    ctx.output(outputTag1, value);
                } else {
                    ctx.output(outputTag2, value);
                }
            }
        });
        DataStream<Integer> evenStream = outputStream.getSideOutput(outputTag1);
        DataStream<Integer> oddStream = outputStream.getSideOutput(outputTag2);

        //evenStream.print().setParallelism(1);

        OutputTag<Integer> outputTag11 = new OutputTag<Integer>("low"){};
        OutputTag<Integer> outputTag22 = new OutputTag<Integer>("high"){};

        //二次切分
        SingleOutputStreamOperator<Integer> subOutputStream = evenStream.process(new ProcessFunction<Integer, Integer>() {
            @Override
            public void processElement(Integer value, Context ctx, Collector<Integer> out) throws Exception {
                if (value <= 5) {
                    ctx.output(outputTag11, value);
                } else {
                    ctx.output(outputTag22, value);
                }
            }
        });

        DataStream<Integer> lowStream = subOutputStream.getSideOutput(outputTag11);
        DataStream<Integer> higtStream = subOutputStream.getSideOutput(outputTag22);
        lowStream.print().setParallelism(1);

        env.execute("StreamSideOutputJava");
    }
}
